#include "thread.h"
#include <iostream>
#include "Log.h"
#include <typeinfo>
#include <message.h>
#include <server.h>
#include "Log.h"
using namespace std;

static int task_timeout = 10;
int task::start(thread_pool *tp)
{
	tp->add_task(this);
	return 0;
}

thread_pool::thread_pool()
{
	_worker = 0;
	_max_number_of_threads = 1;
	_number_of_free_work_threads = 0;
	_number_of_work_threads = 0;
}

thread_pool::~thread_pool()
{
	LOG_DEBUG_VA("~thread_pool() %x, work threads %d", this, _thread_list.size());
	for (list<pthread *>::iterator it = _thread_list.begin(); it != _thread_list.end(); it++) {
		pool_work_thread *_work_thread = static_cast<pool_work_thread *>(*it);
		_work_thread->stop();
		LOG_DEBUG_VA("wait work thread %x end, ", _work_thread);
		_work_thread->wait();
		LOG_DEBUG_VA("work thread %x ended", _work_thread);
		delete *it;
	}

	if (_worker)
		delete _worker;
}

int thread_pool::add_task(task *tsk)
{
	_task_mutex.lock();
	_task_queue.push(tsk);
	LOG_DEBUG_VA("task %x in queue", tsk);
	_task_mutex.unlock();
	if (_number_of_free_work_threads == 0 && _number_of_work_threads < _max_number_of_threads) {
		pool_work_thread *new_work_thread = new pool_work_thread(this);
		_thread_list.push_back(new_work_thread);
		new_work_thread->start();
		_number_of_work_threads++;
		LOG_INFO_VA("new work thread started, total:%d, free:%d", _number_of_work_threads, _number_of_free_work_threads);
	}
	_task_event.signal();
}

int thread_pool::set_worker(worker *wk)
{
	_worker = wk;
	_worker->set_thread_pool(this);
}

int worker::work()
{
	while (!_stoped) {
		task *tsk = _tp->get_task();
		if (!tsk) {
			//_tp->_number_lock.lock();
			//_tp->_number_of_free_work_threads++;
			//_tp->_number_lock.unlock();
			__sync_add_and_fetch(&_tp->_number_of_free_work_threads, 1);
			_tp->_task_event.timedwait(1000);
			// if wait time out
			//	delete this worker;
			__sync_sub_and_fetch(&_tp->_number_of_free_work_threads, 1);
			//_tp->_number_lock.lock();
			//_tp->_number_of_free_work_threads--;
			//_tp->_number_lock.unlock();
		} else {
			msg_handle_task *mt = static_cast<msg_handle_task *>(tsk);
			message *m = mt->get_message();
			socket_event *e = m->get_socket_event();
			timee t = timee::now();
			int interval = t - e->get_active_time();
			LOG_DEBUG_VA("[%s] task %x interval:%d, read interval:%d", e->get_event_manager()->get_server()->whoami().c_str(), tsk, interval, mt->create_time - e->get_active_time());
			if ((timee::now() - e->get_active_time()) < (task_timeout * 1000000))
				tsk->execute(this);
			else {
				LOG_DEBUG_VA("[%s] task %llu-%llu canceled", e->get_event_manager()->get_server()->whoami().c_str(), m->get_head().sn_high, m->get_head().sn_low);
				e->unset_delayed();
			}
			delete tsk;
		}
	}
	LOG_DEBUG_VA("worker stoped work");
}

thread_pool::pool_work_thread::~pool_work_thread()
{
	if (_wkr)
		delete _wkr;
}

task *thread_pool::get_task()
{
	task *t = NULL;
	_task_mutex.lock();
	if (!_task_queue.empty()) {
		t = _task_queue.front();
		_task_queue.pop();
	}
	_task_mutex.unlock();
	LOG_DEBUG_VA("%d tasks in task queue", _task_queue.size());
	LOG_DEBUG_VA("task %x scheduled out", t);
	return t;
}

int thread_pool::pool_work_thread::thread_func()
{
	_wkr = _tp->new_worker();
	_wkr->work();
}

void thread_pool::pool_work_thread::stop()
{
	_wkr->stop();
}

worker *thread_pool::new_worker()
{
	if (!_worker) {
		_worker = new worker();
		_worker->set_thread_pool(this);
	}
	worker *wkr = _worker->clone();
	return wkr;
}
